package db;

import java.util.LinkedList;
import java.util.List;
import java.util.Random;
import java.util.Set;
import java.util.concurrent.atomic.AtomicLong;

import redis.clients.jedis.Response;
import tlsb.DB;
import tlsb.DBException;
import tlsb.Utils;
import util.IPartition;
import util.RedisCluster;
import util.RedisClusterFactory;

public class RedisClientPULL extends DB {
	static AtomicLong midCounter = new AtomicLong(0);
	final private long STARTOFFSET = 0;
	final private long ENDOFFSET = 19;
	final private int MAX = 999999999;
	private static final String SETOFFOLLOWEES = ":fle";
	final private String PREFIXOFUTL = "ut:";

	final private int HTLLENGTH = 20;

	// final private String UIDOFMID = ""
	private final int OperationRetries = 10;

	static Random random = new Random();
	private RedisCluster folRedisCluster = RedisClusterFactory.instance
			.getFolRedisCluster();
	private RedisCluster statusRedisCluster = RedisClusterFactory.instance
			.getStatusRedisCluster();
	private RedisCluster metaDataRedisCluster = RedisClusterFactory.instance
			.getMetaDataRedisCluster();
	private boolean _Debug = false;

	public void init() throws DBException {
	}

	@Override
	public void cleanup() throws DBException {
		folRedisCluster.close();
		statusRedisCluster.close();
		metaDataRedisCluster.close();
	}

	@Override
	public int follow(String uid, String fid) {
		// TODO Auto-generated method stub
		// add to new followers and followers for different strateges; also
		// update owns(first) and belongs
		return 0;
	}

	@Override
	// TODO uid change string into integer.
	public int getHTL(String uid, int start) {
		// get all related stls
		Set<String> fles = folRedisCluster.getJedis(
				folRedisCluster.getPartition().shard(uid)).smembers(
				Utils.joinStr(uid, SETOFFOLLOWEES));// get
		// the
		// list
		// of
		// followers

		List<List<String>> tls = new LinkedList<List<String>>();
		List<Response<List<String>>> responses = new LinkedList<Response<List<String>>>();
		if (fles != null) {
			for (String fle : fles) {
				// stls.add(jedis.lrange(sid, STARTOFFSET, ENDOFFSET));
				int nid = statusRedisCluster.getPartition().shard(fle);
				responses.add(statusRedisCluster.getPipeline(nid)
						.lrange(Utils.joinStr(PREFIXOFUTL, fle), STARTOFFSET,
								ENDOFFSET));
			}

			statusRedisCluster.synAllPipelines();

			for (Response<List<String>> response : responses) {
				tls.add(response.get());
			}
		}
		if (tls != null) {
			RedisClientOVLP.mergeFilter(tls, HTLLENGTH, start);
		}

		if (_Debug) {
			// System.out.println(uid + " PULL:" + htl);
		}
		return 0;
	}

	@Override
	public int newUser() {
		// TODO Auto-generated method stub
		return 0;
	}

	@Override
	public int post(String uid, String mid) {
		statusRedisCluster.getJedis(
				statusRedisCluster.getPartition().shard(uid)).lpush(
				Utils.joinStr(PREFIXOFUTL, uid), mid);
		midCounter.incrementAndGet();
		if (midCounter.get() % 1000000 == 0)
			System.err.println(midCounter.get());
		return 0;
	}

	@Override
	public int unfollow(String uid, String fid) {
		// TODO Auto-generated method stub
		return 0;
	}
}
